package com.xrui.adc.hbase.schema;

import com.google.common.base.Preconditions;

import com.xrui.adc.hbase.codec.RowKeyCodec;
import com.xrui.adc.hbase.hashing.RowKeyHasher;
import com.xrui.adc.hbase.schema.ViewTableSchemaResolver;
import com.xrui.hbase.TableName;
import com.xrui.hbase.conversion.AvroCellConverter;
import com.xrui.hbase.exception.InvalidLayoutException;
import com.xrui.hbase.impl.DefaultHTableInterfaceFactory;
import com.xrui.hbase.schema.*;
import com.xrui.hbase.schema.desc.CellSchemaDesc;
import com.xrui.hbase.schema.desc.ColumnDesc;
import com.xrui.hbase.schema.desc.TableLayoutDesc;
import org.apache.hadoop.conf.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public final class ViewTableLayoutResolver implements TableLayoutResolver {
    private static final Logger LOG = LoggerFactory.getLogger(ViewTableLayoutResolver.class);

    private ViewTableSchemaResolver mSchemaTableResolver;

    public ViewTableSchemaResolver.DerivedSchemaMeta getMeta(TableName tableName, Configuration configuration) {
        TableName schemaTable = TableName.TableNameBuilder.newBuilder()
            .withNamespace(tableName.getNamespace())
            .withPrefix(tableName.getPrefix())
            .withSeparator(tableName.getSeparator())
            .withIdentifier(ViewTableSchemaResolver.SCHEMA_TABLE_NAME)
            .build();
        if (null == mSchemaTableResolver) {
            LOG.debug("Cache miss for schema table resolver");
            try {
                mSchemaTableResolver = new ViewTableSchemaResolver(schemaTable,
                        configuration,
                        DefaultHTableInterfaceFactory.get());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        } else {
            LOG.debug("Cache hit for schema table resolver");
        }
        String schemaID = tableName.toString().replaceAll("_(?:\\d+.)+\\d+", "");
        return mSchemaTableResolver.getMeta(schemaID);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized TableLayout resolve(String tableName, Object resource) throws InvalidLayoutException {
        String schemaID = tableName.replaceAll("_(?:\\d+.)+\\d+", "");
        ViewTableSchemaResolver.DerivedSchemaMeta schemaMeta = getMeta(TableName.valueOf(tableName), (Configuration) resource);
        Preconditions.checkNotNull(schemaMeta, "Not found schema meta for " + schemaID);
        Preconditions.checkNotNull(schemaMeta.getSchema());

        List<RowKeyComponent> rowKeyComponents = new ArrayList<>();
        for (String item : schemaMeta.getHashItems()) {
            rowKeyComponents.add(RowKeyComponent.create(item, RowKeyComponent.ComponentType.STRING, true));
        }
        for (String item : schemaMeta.getUnHashItems()) {
            rowKeyComponents.add(RowKeyComponent.create(item, RowKeyComponent.ComponentType.STRING));
        }
        RowKeyFormat rowKeyFormat = RowKeyFormat.RowKeyFormatBuilder.newBuilder()
            .withDecoder(RowKeyCodec.PROVIDER_NAME)
            .withDelimiter(schemaMeta.getDelimiter())
            .withHasher(RowKeyHasher.PROVIDER_NAME)
            .withEncoding(RowKeyEncoding.FORMATTED)
            .withHashSize(8)
            .withEncoder(RowKeyCodec.PROVIDER_NAME)
            .withComponents(rowKeyComponents)
            .build();
        TableLayoutDesc tableLayoutDesc = TableLayoutDesc.TableLayoutBuilder.newBuilder()
            .withTableName(tableName)
            .withRowKey(rowKeyFormat)
            .addColumn(
                ColumnDesc.ColumnBuilder.newBuilder()
                    .withFamily("cf")
                    .withCellSchema(CellSchemaDesc.CellSchemaBuilder.newBuilder()
                        .withType(SchemaType.AVRO)
                        .withID(schemaID)
                        .withSchema(schemaMeta.getSchema().toString())
                        .withCompression(schemaMeta.getCompression())
                        .withConverter(AvroCellConverter.PROVIDER_NAME)
                        .build()
                    )
                    .build()
            ).build();
        return TableLayout.newLayout(tableLayoutDesc).setSchemaResolver(mSchemaTableResolver);
    }

    @Override
    public String getName() {
        return this.getClass().getName();
    }
}
